package org.kth.dks;

/**
 * <p>Title: DKS</p>
 * <p>Description: DKS Middleware</p>
 * <p>Copyright: Copyright (c) 2005</p>
 * <p>Company: KTH-IMIT/SICS</p>
 * @author Ali Ghodsi (aligh@kth.se)
 * @version 1.0
 */

import java.io.ByteArrayInputStream;
import java.io.ByteArrayOutputStream;
import java.io.IOException;
import java.io.ObjectInputStream;
import java.io.ObjectOutputStream;
import java.util.Collection;
import java.util.Map;
import java.util.Random;
import java.util.Set;

import org.kth.dks.dks_comm.ConnectionManager;
import org.kth.dks.dks_comm.DKSOverlayAddress;
import org.kth.dks.dks_comm.DKSRef;
import org.kth.dks.dks_dht.DKSDHTImpl;
import org.kth.dks.dks_exceptions.DKSIdentifierAlreadyTaken;
import org.kth.dks.dks_exceptions.DKSNodeAlreadyRegistered;
import org.kth.dks.dks_exceptions.DKSRefNoResponse;
import org.kth.dks.dks_exceptions.DKSTooManyRestartJoins;
import org.kth.dks.dks_node.DKSNode;

public class JDHT implements Map
{
  private int PORT = 4440;
  private DKSDHTInterface dht;
  private ConnectionManager cm;

  /**
   * Creates an empty local Distributed Hash Table (DHT), listening on default port 4440
   * Other nodes can connect to this DHT using a reference to this instance
   * An IMPORTANT invariant is to call close() upon exit to disconnect.
   * @see close()
   * @see getReference()
   * @throws IOException thrown if it cannot bind to the local port
   */
  public JDHT() throws IOException {
    constructCreate();
  }

  /**
   * Creates an empty local Distributed Hash Table (DHT), listening on default port 4440
   * Other nodes can connect to this DHT using a reference to this instance
   * An IMPORTANT invariant is to call close() upon exit to disconnect.
   * @see close()
   * @see getReference()
   * @param port int port number to bind to
   * @throws IOException thrown if it cannot bind to the local port
   */
  public JDHT(int port) throws IOException {
    PORT=port;
    constructCreate();
  }

  /**
   * Joins an existing Distributed Hash Table (DHT) running on some machines, will bind to local port 4444
   * An IMPORTANT invariant is to call close() upon exit to disconnect.
   * @param url String reference to an existing node running a DHT
   * @throws IOException thrown if there is a problem binding to the local port
   * @throws DKSTooManyRestartJoins if it cannot join (see DKS documentation)
   * @throws DKSIdentifierAlreadyTaken if the internal DKS identifier is already taken
   * @throws DKSRefNoResponse if the existing node (url) does not respond
   * @see getReference()
   * @see close()
   */
  public JDHT(String url) throws IOException, DKSTooManyRestartJoins, DKSIdentifierAlreadyTaken, DKSRefNoResponse {
    constructJoin(url);
  }

  /**
   * Joins an existing Distributed Hash Table (DHT) running on some machines, will bind to the supplied local port number
   * An IMPORTANT invariant is to call close() upon exit to disconnect.
   * @param port int local port number to bind to
   * @param url String reference to an existing node running a DHT
   * @throws IOException thrown if there is a problem binding to the local port
   * @throws DKSTooManyRestartJoins if it cannot join (see DKS documentation)
   * @throws DKSIdentifierAlreadyTaken if the internal DKS identifier is already taken
   * @throws DKSRefNoResponse if the existing node (url) does not respond
   * @see close()
   * @see getReference()
   */
  public JDHT(int port, String url) throws IOException, DKSTooManyRestartJoins, DKSIdentifierAlreadyTaken, DKSRefNoResponse {
	  PORT=port;
    constructJoin(url);
  }

  private void constructCreate() throws IOException {
      cm = ConnectionManager.getInstance(PORT);
      long id = Math.abs(new Random().nextLong()) % DKSNode.N;
      try {
        dht = new DKSDHTImpl(cm, new DKSOverlayAddress("dksoverlay://0/"+id+"/0"));
        dht.create();
      } catch (DKSNodeAlreadyRegistered ex) {
        cm.end();
        ex.printStackTrace();
      } catch (IOException ex) {
        cm.end();
        throw ex;
      }

  }

  private void constructJoin(String url) throws IOException, DKSTooManyRestartJoins, DKSIdentifierAlreadyTaken, DKSRefNoResponse {
      cm = ConnectionManager.getInstance(PORT);
      long id = Math.abs(new Random().nextLong()) % DKSNode.N;
      try {
        dht = new DKSDHTImpl(cm, new DKSOverlayAddress("dksoverlay://0/"+id+"/0"));
        DKSRef urlref = DKSRef.valueOf(url);
        dht.join(urlref);
      } catch (DKSNodeAlreadyRegistered ex) {
        cm.end();
        ex.printStackTrace();
      } catch (IOException ex) {
        cm.end();
        throw ex;
      }

  }

  /**
   * Inserts an item (key/value pair) into the DHT
   * It requires the value to be serializable, and depends on the hashCode() implementation of key
   * @param key Object key of the item
   * @param value Object value of the item
   * @return Object always null
   */
  public Object put(Object key, Object value)
  {
    ByteArrayOutputStream bout = new ByteArrayOutputStream();
    try {
      ObjectOutputStream out = new ObjectOutputStream(bout);
      out.writeObject(value);
      DKSObject val = new DKSObject(bout.toByteArray());

      // key space is skewed, need to make a real hash using Digest
      dht.addToBinding(key.hashCode(), val);

    } catch (IOException ex) { ex.printStackTrace(); }
    return null;
  }

  /**
   * Gets the value associated with key in the DHT
   * Makes use of hashCode() in key
   * @param key Object key of the item
   * @return Object value associated with key in the DHT, null if none found
   */
  public Object get(Object key)
  {
    DKSObject[] ans = dht.lookupBinding(key.hashCode());
    if (ans.length <=0) return null;

    DKSObject v = ans[0];
    ByteArrayInputStream bout = new ByteArrayInputStream(v.getData());
    try {
      ObjectInputStream in = new ObjectInputStream(bout);
      return in.readObject();
    }
    catch (IOException ex) { ex.printStackTrace(); }
    catch (ClassNotFoundException ex) { ex.printStackTrace(); }
    return null;
  }

  /**
   * Returns a stringified reference to the current DHT instance. This is used to supply other nodes with
   * a reference, such that they can connect to the running DHT
   * @return String a string representing this DHT instance (dksref://hostname:port/gid/id/guid/nonce)
   */
  public String getReference() {
    return dht.getDKSURL();
  }

  /**
   * Unbinds from the local port and disconnects from the DHT. Has to be called before exit.
   */
  public void close() {
    dht.leave();
    try { Thread.sleep(1000); } catch (Exception ex) { ex.printStackTrace(); }
    cm.end();
  }

  /**
   * Returns whether the key exists in the DHT
   * Makes use of hashCode() in key
   * @param key Object key of the item
   * @return boolean is true if key exists in DHT, fales otherwise
   */
  public boolean containsKey(Object key)
  {
    return get(key)==null ? false : true;  // highly inefficient
  }

  /**
   * Currently unimplemented, will throw UnsupportedOperationException
   * @param key Object
   * @throws UnsupportedOperationException
   * @return Object
   */
  public Object remove(Object key) // throws UnsupportedOperationException
  {
	  //get(key.hashCode());
	  DKSObject[] ans = dht.lookupBinding(key.hashCode());
	  if (ans.length <=0) return null;

	  dht.removeFromBinding(key.hashCode(),ans[0]);

	  //DKSObject v = ans[0];
	  ByteArrayInputStream bout = new ByteArrayInputStream(ans[0].getData());
	  try {
		  ObjectInputStream in = new ObjectInputStream(bout);
		  return in.readObject();
	  }
	  catch (IOException ex) { ex.printStackTrace(); }
	  catch (ClassNotFoundException ex) { ex.printStackTrace(); }
	  return null;
  }

  /**
   * Currently unimplemented, will throw UnsupportedOperationException
   * @throws UnsupportedOperationException
   */
  public void clear()//  throws UnsupportedOperationException
  { 
	  dht = null;
	  System.gc();
	  //throw new UnsupportedOperationException();
  }

  /**
   * Currently unimplemented, will throw UnsupportedOperationException
   * @param value Object
   * @throws UnsupportedOperationException
   * @return boolean
   */
  public boolean containsValue(Object value)  throws UnsupportedOperationException
  { throw new UnsupportedOperationException(); }

  /**
   * Currently unimplemented, will throw UnsupportedOperationException
   * @throws UnsupportedOperationException
   * @return Set
   */
  public Set entrySet() throws UnsupportedOperationException
  { throw new UnsupportedOperationException(); }

  /**
   * Currently unimplemented, will throw UnsupportedOperationException
   * @param o Object
   * @throws UnsupportedOperationException
   * @return boolean
   */
  public boolean equals(Object o)  throws UnsupportedOperationException
  { throw new UnsupportedOperationException(); }

  /**
   * Currently unimplemented, will throw UnsupportedOperationException
   * @throws UnsupportedOperationException
   * @return boolean
   */
  public boolean isEmpty()  throws UnsupportedOperationException
  { throw new UnsupportedOperationException(); }

  /**
   * Currently unimplemented, will throw UnsupportedOperationException
   * @throws UnsupportedOperationException
   * @return Set
   */
  public Set keySet()  throws UnsupportedOperationException
  { throw new UnsupportedOperationException(); }

  /**
   * Currently unimplemented, will throw UnsupportedOperationException
   * @param t Map
   * @throws UnsupportedOperationException
   */
  public void putAll(Map t)  throws UnsupportedOperationException
  { throw new UnsupportedOperationException(); }

  /**
   * Currently unimplemented, will throw UnsupportedOperationException
   * @throws UnsupportedOperationException
   * @return int
   */
  public int size()  throws UnsupportedOperationException
  { throw new UnsupportedOperationException(); }

  /**
   * Currently unimplemented, will throw UnsupportedOperationException
   * @throws UnsupportedOperationException
   * @return Collection
   */
  public Collection values()  throws UnsupportedOperationException
  { throw new UnsupportedOperationException(); }


}
